INFOPLAT 3099 chip ingress batching #1756
Conversation
hendoxc
commented
Jan 5, 2026
hendoxc
commented
- Adds chipingress batching client
- Adds unit-tests
product-security-plaid-production
bot
requested review from
4of9,
engnke,
fouadkada,
kirqz23,
pkcll and
thomaska
|
👋 hendoxc, thanks for creating this pull request! To help reviewers, please consider creating future PRs as drafts first. This allows you to self-review and make any final changes before notifying the team. Once you're ready, you can mark it as "Ready for review" to request feedback. Thanks! |
There was a problem hiding this comment.
Pull request overview
This PR implements a batching client for chipingress that collects CloudEvent messages and sends them in batches based on configurable size and timeout triggers. The implementation provides non-blocking message queuing with configurable concurrency control.
Key changes:
- Implements a
BatchClientwith configurable batch size, timeout, buffer size, and concurrency limits - Provides asynchronous batch processing with proper context cancellation and shutdown handling
- Includes comprehensive unit tests covering batch triggers, edge cases, and concurrent operations
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| pkg/chipingress/batch/client.go | Implements the core BatchClient with Start/Stop lifecycle, message queuing, and batch sending logic with configurable options |
| pkg/chipingress/batch/client_test.go | Provides comprehensive test coverage for client initialization, message queuing, batch sending, and various trigger scenarios |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
✅ API Diff Results - No breaking changes |
FIxes test
hendoxc
force-pushed
the
INFOPLAT-3099-chip-ingress-batching
branch
from
e705b9f to
703498f
Compare
- ensures no panic if .Stop is called multiple times
…tcontractkit/chainlink-common into INFOPLAT-3099-chip-ingress-batching
| result := make([]*messageWithCallback, len(b.messages)) | ||
| copy(result, b.messages) |
There was a problem hiding this comment.
Could clone here:
| result := make([]*messageWithCallback, len(b.messages)) | |
| copy(result, b.messages) | |
| result := slices.Clone(b.messages) |
| copy(result, b.messages) | ||
|
|
||
| // Reset the internal slice | ||
| b.messages = b.messages[:0] |
There was a problem hiding this comment.
Do we need to consider freeing this memory in some cases so an outlier capacity doesn't leave a large buffer in the background indefinitely?
| result := make([]*messageWithCallback, len(b.messages)) | ||
| copy(result, b.messages) | ||
| return result |
There was a problem hiding this comment.
| result := make([]*messageWithCallback, len(b.messages)) | |
| copy(result, b.messages) | |
| return result | |
| return slices.Clone(b.messages) |
| batchInterval time.Duration | ||
| maxPublishTimeout time.Duration | ||
| messageBuffer chan *messageWithCallback | ||
| shutdownChan chan struct{} |
There was a problem hiding this comment.
If you use a services.StopChan, then you can derive contexts from it instead of using context.Background():
| shutdownChan chan struct{} | |
| shutdownChan services.StopChan |
4 participants
